package com.at.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/29 9:24 PM
 */
object SparkStreaming02_Queue {

  def main(args: Array[String]): Unit = {
    // 创建环境
    // 创建时，需要传递2个参数：
    //    param： 环境配置
    val sc = new SparkConf().setMaster("local[*]").setAppName("operator")
    //    param2: 采集周期
    val ssc = new StreamingContext(sc, Seconds(3))

    val rddQueue = new mutable.Queue[RDD[Int]]()

    val inputStream = ssc.queueStream(rddQueue, oneAtATime = false)
    val mappedStream = inputStream.map((_, 1))
    val reduceStream = mappedStream.reduceByKey(_ + _)
    reduceStream.print()

    // 1、启动采集器
    ssc.start()
    for (i <- 1 to 5) {
      rddQueue += ssc.sparkContext.makeRDD(1 to 300, numSlices = 10)
      Thread.sleep(2000)
    }
    // 2、等待采集器的关闭
    ssc.awaitTermination()
  }
}
